package com.taoqi.ss

import java.util.regex.Pattern

import org.apache.spark.sql.types.{StringType, StructField, StructType}

import scala.collection.mutable.ArrayBuffer

/**
  * Created by TQ on 2017/3/30.
  */
object JsonSchemaBuilder {

  final val columnSplitPattern = Pattern.compile("\\s*,\\s*")
  private final val fieldSplitPattern = Pattern.compile("\\.")
  private final val fieldPattern = Pattern.compile("([\\w\\.]+)(?:\\s+as\\s+\\w+)?")

  def getJsonSchema(schema: String): StructType = {
    getSchemaByFieldsList(columnSplitPattern.split(schema).map(getFieldList).toList)
  }

  private def getFieldList(singleField: String): List[String] = {
    val fieldMatch = fieldPattern.matcher(singleField)
    if (fieldMatch.matches()) {
      val fieldSource = fieldMatch.group(1)
      val fieldArray = fieldSplitPattern.split(fieldSource)
      fieldArray.toList
    } else {
      throw new IllegalArgumentException(s"field format error:$singleField ,we need parent.children(as aliasName)")
    }
  }

  private def getSchemaByFieldsList(fieldsList: List[List[String]]): StructType = {
    fieldsList.map(getStrcutType).reduce(mergeStructType)
  }

  private def getStrcutType(fields: List[String]): StructType = {
    fields match {
      case head :: Nil => StructType(StructField(head, StringType, true) :: Nil)
      case head :: tail => StructType(StructField(head, getStrcutType(tail), true) :: Nil)
      case _ =>  StructType(Nil)
    }
  }

  private def mergeStructType(left: StructType, right: StructType): StructType = {
    val newFields = ArrayBuffer.empty[StructField]
    val leftFields = left.fields
    val rightFields = right.fields
    val rightMapped = fieldsMap(rightFields)
    leftFields.foreach {
      case leftField@StructField(leftName, leftType, leftNullable, _) =>
        rightMapped.get(leftName)
          .map {
            case rightField@StructField(_, rightType, rightNullable, _) =>
              leftField.copy(
                dataType = mergeStructType(leftType.asInstanceOf[StructType], rightType.asInstanceOf[StructType]),
                nullable = leftNullable || rightNullable)
          }
          .orElse(Some(leftField))
          .foreach(newFields += _)
    }

    val leftMapped = fieldsMap(leftFields)
    rightFields
      .filterNot(f => leftMapped.get(f.name).nonEmpty)
      .foreach(newFields += _)
    StructType(newFields)
  }

  private def fieldsMap(fields: Array[StructField]): Map[String, StructField] = {
    import scala.collection.breakOut
    fields.map(s => (s.name, s))(breakOut)
  }

  def main(args: Array[String]): Unit = {

    /*GpsInfo(imei: String,gpsTime: String,receiveTime: String,lng: String,lat: String,
      course: String,speed: String, gsmSignal: String, acc: String, targetingStatus: String,
      oilStatus: String,speedAlarm: String,vibrationAlarm: String,powerAlarm: String,
      cellId: String,lac: String,voltage: String,gpsSignal: String,ip: String,
      port: String,electricity: String)*/
    /*val schema ="props,\n event.info.eventId,\n props.age,\n event.own.telnum as tel"*/
    val schemaString = "imei,\n gpsTime,\n receiveTime,\n lng,\n lat,\n course,\n speed,\n gsmSignal," +
      "\n acc,\n targetingStatus,\n oilStatus,\n speedAlarm,\n vibrationAlarm,\n powerAlarm," +
      "\n cellId,\n lac,\n voltage,\n gpsSignal,\n electricity,\n gpsStatus,\n latStatus,\n lngStatus," +
      "\n powerStatus,\n sosStatus,\n workStatus,\n status,\n mcc,\n mnc,\n workStyle,\n sim,\n currentTime," +
      "\n province,\n city"

    val se = """
               | FCustomer \n
             """.stripMargin
    JsonSchemaBuilder.getJsonSchema(schemaString).printTreeString()

    /*sqlContext.read.schema(JsonSchemaBuilder.getJsonSchema(schemaString)).load(path)
      .selectExpr(JsonSchemaBuilder.columnSplitPattern.split(schemaString):_*)*/
  }
}